package com.baishancloud.log.traffic.jh

import com.baishancloud.log.common.util.other.DataTimeFilterUtil
import com.baishancloud.log.format.scala.Origin
import com.baishancloud.log.traffic.{filterDomains, filterSecond}
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration
import org.slf4j.{Logger, LoggerFactory}

import java.util.Objects

/**
 * 对指定的域名进行最大时间过滤，该过滤适用于金华地区运行的任务
 *
 * @author ziqiang.wang 
 * @date 2022/3/31 17:56
 */
class DataFilterJh(parameterTool: ParameterTool) extends RichFilterFunction[Origin] {

  var filterDomainsSet: Set[String] = _
  var maxSecond: Long = Long.MaxValue

  override def open(parameters: Configuration): Unit = {
    filterDomainsSet = parameterTool.get(filterDomains).split(",").toSet
    maxSecond = parameterTool.getLong(filterSecond)
  }

  /**
   * 在原先的基础上，增加对指定域名的时间过滤，该类对应任务运行在金华地区服务器，指定时间为消费的最大时间，判断条件为 <
   */
  override def filter(value: Origin): Boolean = {
    Objects.equals(value.xPeer(), "-") &&
      value.url() != null &&
      value.url().contains("bsxdisp=se") &&
      DataTimeFilterUtil.domainSecondFilterMax(value, filterDomainsSet, maxSecond)
  }
}
